Just a style not about the test. The implementation of this explicit timeout is usually not necessary. The gen_cluster ships with a timeout. I believe this pattern originated before the gen_cluster had its own timeout.

Thanks for the historical context, that's good to know. I'll take these out, though I will also make the gen_cluster timeout shorter—30s is a long time to wait for these to fail. If everything's working, the cluster shuts down instantly, so 5-10sec is plenty to know that something's wrong.

though I will also make the gen_cluster timeout shorter—30s is a long time to wait for these to fail.

It is typically that large because on CI the runtimes vary a lot. For local testing I usually also adapt this. FWIW, there is a constant in utils_test which controls all known timeouts (gen_cluster and gen_test) so you may change that for local testing as well. If it helps we could also allow this to be set with an environment variable to make it easier for local dev.

Note, though, I believe the other tests are not protected by a timeout by default. We only have the global pytest-timeout last resort timeout

bottom line: you may keep your timeout, of course, I just wanted to fill in some info

I would not advise to reduce timeout below 30s. It's very easy to set something that looks reasonably comfortable on your laptop, only to find out later that CI can get 10x slower.

why is the inform not always necessary?

This is the key change in this PR. See #4936 (particularly the "complex case") for a description of the case this is solving. Basically, when an Actor handle got serialized and transferred to another worker, that new worker was taking out a lease on the Actor's key, which prevented that key from ever being released because the scheduler saw that some client (that worker) wanted it, which meant the scheduler never told that worker to release the key. In the end, the Actor would continue running even when nothing depended on it.

But also as I mentioned in the description, I don't think this is a good implementation. I just wanted to post the problem and a quick fix, and mostly hear from @mrocklin on what the worker= kwarg to Actor was supposed to be used for (since I can't find it used anywhere).

I've been thinking something like

 diff --git a/distributed/actor.py b/distributed/actor.py
index 77b2cda6..79e96f5b 100644
--- a/distributed/actor.py
+++ b/distributed/actor.py
@@ -49,24 +49,22 @@ class Actor(WrappedKey):
     2
     """
 
-    def __init__(self, cls, address, key, worker=None):
+    def __init__(self, cls, address, key):
         self._cls = cls
         self._address = address
         self.key = key
         self._future = None
-        if worker:
-            self._worker = worker
-            self._client = None
-        else:
-            try:
-                self._worker = get_worker()
-            except ValueError:
-                self._worker = None
+        self._client = None
+        self._worker = None
+
+        try:
+            self._worker = get_worker()
+        except ValueError:
             try:
-                self._client = default_client()
+                self._client = get_client()
                 self._future = Future(key)
             except ValueError:
-                self._client = None
+                pass
 
     def __repr__(self):
         return "<Actor: %s, key=%s>" % (self._cls.__name__, self.key)

where we remove the worker= argument, and make self._client and self._worker mutually exclusive. This way, whenever we're running on a worker, we don't create a Future.

If the Actor handle was sent to another worker because a task depends on that Actor's key, holding the Future is unnecessary—the fact that a task depends on that key means the scheduler won't release that key yet. The only issue could be running a task where the function happens to connect to an Actor internally (but the task has no explicit dependency on that Actor)—in that case, if the client released the Actor's key while that task was running, in theory the Actor could be cancelled while that task needed it, since the task doesn't itself hold a Future to the key (since get_worker() succeeded).

That sounds a bit like the worker where the actor is deserialized is not doing a good job of cleaning up after itself.
Worker A: owner of the actor; serialize
Worker B: interested in actor result; deserialize

In particular, once the key/actor is removed on Worker B (think release_key), the future should be released. Once the future is released, the scheduler should trigger the proper forget lifecycle for the actor.

Inspecting the actor class, we do not have a dedicated release mechanism. Therefore, the future is coupled to the lifetime of the client the future is attached to. The client will probably live as long as the worker lives and therefore will only be cleaned up once the worker closes. Even worse in same-process worker situations where clients may be shared between workers, the future may even outlive the worker which cause the reference.

I argue that something like the following should do the trick

diff --git a/distributed/actor.py b/distributed/actor.py
index 231cc8b3..b40e4c1a 100644
--- a/distributed/actor.py
+++ b/distributed/actor.py
@@ -64,10 +64,18 @@ class Actor(WrappedKey):
                 self._worker = None
             try:
                 self._client = get_client()
-                self._future = Future(key, inform=self._worker is None)
+                self._future = Future(key)
             except ValueError:
                 self._client = None

+    def release(self):
+        if self._future:
+            self._future.release()
+            self._future = None
+
+    def __del__(self):
+        self.release()
+
     def __repr__(self):
         return "<Actor: %s, key=%s>" % (self._cls.__name__, self.key)


diff --git a/distributed/worker.py b/distributed/worker.py
index 44e05f05..27c9de97 100644
--- a/distributed/worker.py
+++ b/distributed/worker.py
@@ -2643,11 +2643,13 @@ class Worker(ServerNode):
                 self.log.append((key, "release-key", {"cause": cause}, reason))
             else:
                 self.log.append((key, "release-key", reason))
+
             if key in self.data and not ts.dependents:
-                try:
-                    del self.data[key]
-                except FileNotFoundError:
-                    logger.error("Tried to delete %s but no file found", exc_info=True)
+                data = self.data.pop(key)
+                from distributed.actor import Actor
+                if isinstance(data, Actor):
+                    data.release()
+                del data
             if key in self.actors and not ts.dependents:
                 del self.actors[key]

however, this also deadlocks and doesn't release. this deadlock seems to connect to the recent worker state machine issues and loosely connects to #4918 I'm not entirely sure if the fixes over there would resolve that deadlock, though. To get the tests unstuck I needed to add one more patch.

FWIW, I think your approach is fine. For ordinary data, workers are not able to hold references to remote data and block their release. why should it be any different for actors?

Worker A: owner of the actor; serialize
Worker B: interested in actor result; deserialize

In particular, once the key/actor is removed on Worker B (think release_key), the future should be released.

The issue is that there may be nothing to cause the key/actor to be removed on Worker B—release_key will never run.

Let's call the key actor-abcde. The issue is that worker B holds a future for actor-abcde. When everything that depends on actor-abcde completes, _propagate_forgotten runs on the scheduler. It sees that actor-abcde has no dependents. But it also sees that TaskState('actor-abcde').who_wants is not empty.

For ordinary data, workers are not able to hold references to remote data and block their release. why should it be any different for actors?

Good point. That makes me feel comfortable going to go with the approach I showed above.

Is there a test showing that an actor is not currently cleaned up?

If this is running in a task (normal or actor), then the thread-local execution state must contain the right worker, no await would have happened, no? You're thinking of cases where the task itself is async?

See this line in my related unmerged PR.

Is there a test showing that an actor is not currently cleaned up?